package yunjiao.javatutorials.guava.concurrent.event;

import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * 事件总线集成
 *
 * @author yangyunjiao
 */
public class EventProcessor {
    private final Map<String, SettableFuture<String>> pendingRequests = new ConcurrentHashMap<>();

    /**
     * 发送请求并返回 Future
     * @param requestId 请求ID
     * @param message 请求消息
     * @return 异步结果 Future
     */
    public ListenableFuture<String> sendRequest(String requestId, String message) {
        SettableFuture<String> future = SettableFuture.create();
        pendingRequests.put(requestId, future);

        // 发送消息到消息队列
        sendToQueue(requestId, message);

        System.out.println("发送请求: " + requestId + ", 消息: " + message);
        return future;
    }

    /**
     * 处理成功响应
     * @param requestId 请求ID
     * @param response 响应内容
     */
    public void handleResponse(String requestId, String response) {
        SettableFuture<String> future = pendingRequests.remove(requestId);
        if (future != null) {
            System.out.println("处理响应: " + requestId + ", 响应: " + response);
            future.set(response);
        } else {
            System.out.println("未找到对应的请求: " + requestId);
        }
    }

    /**
     * 处理错误响应
     * @param requestId 请求ID
     * @param error 错误信息
     */
    public void handleError(String requestId, String error) {
        SettableFuture<String> future = pendingRequests.remove(requestId);
        if (future != null) {
            System.out.println("处理错误: " + requestId + ", 错误: " + error);
            future.setException(new RuntimeException(error));
        } else {
            System.out.println("未找到对应的请求: " + requestId);
        }
    }

    /**
     * 模拟发送消息到消息队列
     * @param requestId 请求ID
     * @param message 消息内容
     */
    private void sendToQueue(String requestId, String message) {
        // 这里模拟发送到消息队列的实际逻辑
        // 在实际应用中，这里可能是 RabbitMQ、Kafka 等消息中间件
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println("消息已发送到队列 - 请求ID: " + requestId);
    }

    /**
     * 获取待处理请求数量
     * @return 待处理请求数
     */
    public int getPendingRequestCount() {
        return pendingRequests.size();
    }

    /**
     * 清理指定请求
     * @param requestId 请求ID
     */
    public void cleanupRequest(String requestId) {
        SettableFuture<String> future = pendingRequests.remove(requestId);
        if (future != null && !future.isDone()) {
            future.cancel(true);
            System.out.println("已清理请求: " + requestId);
        }
    }
}
